package net.lesscoding.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

import static net.lesscoding.kafka.common.Const.OS_KAFKA;

/**
 * @author eleven
 * @date 2024/6/29 15:52
 * @apiNote
 */
public class KafkaProducerInterceptorTest {
    public static void main(String[] args) {
        // 1. 创建生产者对象
        // 1.1 创建配置对象
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, OS_KAFKA);
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ValueInterceptorTest.class);
        // 服务地址
        StringSerializer stringSerializer = new StringSerializer();
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties,
                stringSerializer,
                stringSerializer);
        // 2. 创建数据
        /**
         * 1. topic名称： 主题不存在则会自动创建
         * 2. 数据的KEy
         * 3. 数据的value
         */
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(
                    "test",
                    "key" + i,
                    "value" + i
            );
            // 3. 通过生产者将数据发送到Kafka
            producer.send(record);
        }

        // 4. 关闭生产者对象
        producer.close();
    }
}
